'''
向topic中发送数据，使用命令：
python ./kafka_avro_send_one_data.py --bootstrap_servers 172.16.80.10:6667,172.16.80.11:6667 --topic_name deye_dwd_pr_location --json_path E:/python_code_using_pycharm/little_tools/used_file/test_json.json --schema_path E:/DEYE-6.2/DEYE-6.2.3/schema/dwd/dwd-pr-location.avsc
bootstrap_servers   kafka节点
topic_name          topic名称
json_path           json文件路径，json文件中写需要发送的数据，当数据为dict时，作为一条数据发送；当数据为list时，作为多条数据发送
schema_path         schema文件的路径

'''

import io,sys
import getopt
import avro
from avro.io import DatumWriter
from kafka import SimpleProducer
from kafka import KafkaClient
import json


def send_data(bootstrap_servers,topic_name,json_path,SCHEMA_PATH):
    json_file = open(json_path,"r",encoding="utf-8")
    dic_data = json.load(json_file)

    # To send messages synchronously
    KAFKA = KafkaClient(bootstrap_servers)
    # Path to user.avsc avro schema
    PRODUCER = SimpleProducer(KAFKA)
    SCHEMA = avro.schema.parse(open(SCHEMA_PATH).read())

    if type(dic_data) == dict:
        dic_data = [dic_data]

    for data in dic_data:
        writer = DatumWriter(SCHEMA)
        bytes_writer = io.BytesIO()
        encoder = avro.io.BinaryEncoder(bytes_writer)
        writer.write(data, encoder)
        raw_bytes = bytes_writer.getvalue()
        PRODUCER.send_messages(topic_name, raw_bytes)



def main(argv):
    try:
        opts, args = getopt.getopt(argv, "b:t:j:s", ["bootstrap_servers=", "topic_name=", "json_path=", "schema_path="])
    except:
        print("input is wrong ! \nexample: python ./kafka_avro_send_one_data.py --bootstrap_servers 172.16.80.10:6667,172.16.80.11:6667 --topic_name deye_dwd_pr_location --json_path E:/python_code_using_pycharm/little_tools/test_json.json --schema_path E:/DEYE-6.2/DEYE-6.2.3/schema/dwd/dwd-pr-location.avsc")
        exit(2)
    for opt, arg in opts:
        if opt in ("-b", "--bootstrap_servers"):
            bootstrap_servers = arg
        if opt in ("-t", "--topic_name"):
            topic_name = arg
        if opt in ("-j", "--json_path"):
            json_path = arg
        if opt in ("-s", "--schema_path"):
            schema_path = arg
    send_data(bootstrap_servers,topic_name,json_path,schema_path)


if __name__ == '__main__':
    main(sys.argv[1:])
